package com.rem.flink.flink5Watermark;

import com.rem.flink.flink2Source.ClickSource;
import com.rem.flink.flink2Source.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 从输入端获取
 * 无界流处理  针对乱序流 forBoundedOutOfOrderness
 *
 * @author Rem
 * @date 2022-09-26
 */

public class WaterMarkBoundedOutTest {
    public static void main(String[] args) throws Exception {
        //1.创建执行环境
        StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.从socket中读取流
        DataStreamSource<Event> stream = streamEnv.addSource(new ClickSource());

        // 插入水位线的逻辑
        stream.assignTimestampsAndWatermarks(
                        //插入水位线 针对乱序设置延迟时间  设置延迟时间为0 就和顺序流是一致的
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                                //提取时间戳  取事件事件
                                .withTimestampAssigner((element, recordTimestamp) -> element.getTimestamp()))
                .keyBy(Event::getUser)
                //设置滑动事件窗口
                .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                // 自定义窗口处理函数
                .process(new WatermarkTestResult())
                .print();

        //7.执行流
        streamEnv.execute();

    }


    /**
     * 获得context 获取时间窗口
     * 自定义处理窗口函数，输出当前的水位线和窗口信息
     */
    public static class WatermarkTestResult extends ProcessWindowFunction<Event, String, String, TimeWindow> {
        @Override
        public void process(String s, Context context, Iterable<Event> elements, Collector<String> out) {
            long start = context.window().getStart();
            long end = context.window().getEnd();
            long currentWatermark = context.currentWatermark();
            long count = elements.spliterator().getExactSizeIfKnown();
            out.collect("窗口" + start + " ~ " + end + "中共有" + count + "个元素，窗口闭合计算时，水位线处于：" + currentWatermark);
        }
    }
}
